package cn.wangjie.spark.work

import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 基于IDEA集成开发环境，编程实现：从TCP Socket实时读取流式数据，对每批次中数据进行词频统计WordCount。
 */
object StreamingBlockDStream {
	
	def main(args: Array[String]): Unit = {
		
		// TODO: 1. 创建StreamingContext流式上下文实例对象，设置BatchInterval
		val ssc: StreamingContext = {
			// 1.a 创建SparkConf对象，设置应用名称和运行模式
			val sparkConf: SparkConf = new SparkConf()
    			.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
    			.setMaster("local[3]")
			// 1.b 传递SparkConf和设置BatchInterval构建流式上下文对象
			new StreamingContext(sparkConf, Seconds(1))
		}
		
		// TODO: 2. 从数据源实时消费数据，此处为TCP Socket
		/*
		  def socketTextStream(
		      hostname: String,
		      port: Int,
		      storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
		    ): ReceiverInputDStream[String]
		 */
		val inputDStream: DStream[String] = ssc.socketTextStream(
			"",
			9999,
			storageLevel = StorageLevel.MEMORY_AND_DISK
		)
		
		// TODO: 3. 实时对每批次数据进行处理
		val resultDStream: DStream[(String, Int)] = inputDStream
			// 将每行数据按照分隔符划分
			.flatMap(line => line.trim.split("\\s+"))
			// 每个单词转换为二元组，表示单词出现一次
			.map(word => word -> 1)
			// 按照单词分组聚合统计
			.reduceByKey((tmp, item) => tmp + item)
		
		// TODO: 4. 每批次处理结果数据进行输出
		resultDStream.print()
		
		// TODO: 5. 启动流式应用
		ssc.start()  // 当流式应用启动以后，将会一直运行，除非认为终止或者程序异常终止
		ssc.awaitTermination()
		// 当程序终止以后，结束应用
		ssc.stop(stopSparkContext = true, stopGracefully = true)
	}
	
}
